// 文件路径: dbms/execution/execution/InsertExecutor.java
package com.dbms.execution.execution;

import com.dbms.catalog.Catalog;
import com.dbms.common.RID;
import com.dbms.common.Schema;
import com.dbms.common.Tuple;
import com.dbms.common.Type;
import com.dbms.execution.plan.PlanNode;
import com.dbms.storage.BufferPoolManager;
import com.dbms.storage.Page;
import com.dbms.storage.TablePage;
import com.dbms.transaction.LogManager;
import com.dbms.transaction.LogRecord;
import com.dbms.transaction.Transaction;

import java.io.IOException;
import java.util.List;
import java.util.Optional;

/**
 * 【已修改】
 * InsertExecutor (插入执行器) 负责处理元组的插入操作。
 * 这个版本在插入数据前，增加了主键约束、VARCHAR长度约束和外键约束的校验，
 * 以确保数据的完整性和一致性。
 */
public class InsertExecutor implements Executor {
    private final BufferPoolManager bufferPoolManager;
    private final Catalog.TableMetadata tableInfo;
    private final List<Tuple> tuplesToInsert;
    private boolean hasExecuted = false;
    private final Transaction transaction;
    private final LogManager logManager;
    private final Catalog catalog;

    /**
     * 完整的构造函数。
     * @param bpm 缓冲池管理器
     * @param tableInfo 目标表的元数据
     * @param tuples 要插入的元组列表
     * @param transaction 事务上下文
     * @param logManager 日志管理器
     * @param catalog 目录管理器（用于外键检查）
     */
    public InsertExecutor(BufferPoolManager bpm, Catalog.TableMetadata tableInfo, List<Tuple> tuples, Transaction transaction, LogManager logManager, Catalog catalog) {
        this.bufferPoolManager = bpm;
        this.tableInfo = tableInfo;
        this.tuplesToInsert = tuples;
        this.transaction = transaction;
        this.logManager = logManager;
        this.catalog = catalog;
    }

    /**
     * 兼容旧代码的构造函数。
     */
    public InsertExecutor(BufferPoolManager bpm, Catalog.TableMetadata tableInfo, List<Tuple> tuples) {
        this(bpm, tableInfo, tuples, null, null, null);
    }

    @Override
    public void open() { hasExecuted = false; }

    /**
     * 执行插入操作。
     * @return 总是返回 null，因为 DML 操作不返回元组。
     */
    @Override
    public Tuple next() {
        if (hasExecuted || tuplesToInsert.isEmpty()) return null;
        if (tableInfo == null) throw new IllegalStateException("Table does not exist.");
        try {
            for (Tuple tuple : tuplesToInsert) {
                // 1. 校验元组数据是否符合Schema定义（如VARCHAR长度）
                validateTuple(tuple);
                // 2. 检查外键约束
                checkForeignKeyConstraint(tuple);
                // 3. 检查主键唯一性约束
                checkPrimaryKeyConstraint(tuple);
                // 4. 执行插入并记录日志
                if (insertTupleAndLog(tuple) == null) {
                    System.err.println("Failed to insert tuple, stopping execution.");
                    break;
                }
            }
        } catch (RuntimeException | IOException e) {
            // 捕获并重新抛出任何在校验或插入过程中发生的异常
            throw new RuntimeException(e);
        }
        hasExecuted = true;
        return null;
    }

    /**
     * 检查待插入元组是否违反外键约束。
     * @param tuple 待插入的元组
     * @throws IOException 如果索引查找失败
     */
    private void checkForeignKeyConstraint(Tuple tuple) throws IOException {
        // 遍历当前表的所有外键定义
        for (Catalog.ForeignKeyInfo fk : tableInfo.foreignKeys) {
            int localColumnIndex = -1;
            // 找到本表中外键列的索引
            for (int i = 0; i < tableInfo.schema.getColumnCount(); i++) {
                if (tableInfo.schema.getColumn(i).getColumnName().equalsIgnoreCase(fk.sourceColumnName)) {
                    localColumnIndex = i;
                    break;
                }
            }

            if (localColumnIndex != -1) {
                // 获取待插入元组中的外键值
                Object foreignKeyValue = tuple.getValue(localColumnIndex);
                // 获取被引用的表（父表）的元数据
                Catalog.TableMetadata referencedTable = catalog.getTable(fk.referencedTableName);
                // 使用父表的索引来查找这个外键值是否存在
                if (referencedTable.index.find((Integer) foreignKeyValue) == null) {
                    // 如果在父表中找不到，则违反了外键约束，抛出异常
                    throw new RuntimeException("Foreign key constraint violation: value " + foreignKeyValue + " not found in " + fk.referencedTableName);
                }
            }
        }
    }

    /**
     * 验证元组中的值是否符合表的 Schema 定义，特别是 VARCHAR 的长度。
     * @param tuple 待验证的元组
     */
    private void validateTuple(Tuple tuple) {
        for (int i = 0; i < tableInfo.schema.getColumnCount(); i++) {
            com.dbms.common.Column col = tableInfo.schema.getColumn(i);
            if (col.getColumnType() == Type.VARCHAR) {
                if (col.getVarcharLength() > 0) {
                    String value = (String) tuple.getValue(i);
                    if (value.length() > col.getVarcharLength()) {
                        throw new RuntimeException(
                                String.format("Value too long for column '%s' (max: %d, actual: %d)",
                                        col.getColumnName(), col.getVarcharLength(), value.length())
                        );
                    }
                }
            }
        }
    }

    /**
     * 检查待插入元组是否违反主键的唯一性约束。
     * @param tuple 待插入的元组
     * @throws IOException 如果索引查找失败
     */
    private void checkPrimaryKeyConstraint(Tuple tuple) throws IOException {
        // 找到主键列
        Optional<com.dbms.common.Column> pkColumnOpt = tableInfo.schema.getColumns().stream()
                .filter(com.dbms.common.Column::isPrimaryKey).findFirst();
        // 如果没有主键或主键不是INTEGER类型（当前索引只支持INTEGER），则跳过检查
        if (pkColumnOpt.isEmpty() || pkColumnOpt.get().getColumnType() != Type.INTEGER) return;

        int pkIndex = tableInfo.schema.getColumns().indexOf(pkColumnOpt.get());
        // 使用索引查找该主键值是否已存在
        if (tableInfo.index.find((Integer) tuple.getValue(pkIndex)) != null) {
            // 如果存在，则违反了主键约束，抛出异常
            throw new RuntimeException("PRIMARY KEY constraint violation: Duplicate key value is (" + tuple.getValue(pkIndex) + ")");
        }
    }

    /**
     * 将元组插入到表页面中，并记录相应的日志和索引。
     * @param tuple 要插入的元组
     * @return 插入成功后元组的 RID，失败则返回 null。
     * @throws IOException
     */
    private RID insertTupleAndLog(Tuple tuple) throws IOException {
        int currentPageId = tableInfo.rootPageId;
        RID rid = null;
        while (true) {
            Page currentPage = bufferPoolManager.fetchPage(currentPageId);
            if (currentPage == null) return null;
            currentPage.wLock(); // 加写锁
            try {
                TablePage currentPageView = new TablePage(currentPage);
                rid = currentPageView.insertTuple(tuple.getData());
                if (rid != null) {
                    // 插入成功，解钉脏页并跳出循环
                    bufferPoolManager.unpinPage(currentPageId, true);
                    break;
                }
                // 当前页面已满，尝试移动到下一页
                int nextPageId = currentPageView.getNextPageId();
                if (nextPageId != TablePage.INVALID_PAGE_ID) {
                    bufferPoolManager.unpinPage(currentPageId, false);
                    currentPageId = nextPageId;
                } else {
                    // 没有下一页，需要创建一个新页面
                    Page newPage = bufferPoolManager.newPage();
                    if (newPage == null) {
                        bufferPoolManager.unpinPage(currentPageId, false);
                        return null; // 无法创建新页面
                    }
                    System.out.println("    [演示信息] 页面 " + currentPageId + " 已满，创建并链接新页面 " + newPage.getPageId());
                    newPage.wLock();
                    try {
                        TablePage newPageView = new TablePage(newPage);
                        newPageView.init();
                        rid = newPageView.insertTuple(tuple.getData());
                        currentPageView.setNextPageId(newPage.getPageId()); // 将新页面链接到旧页面后面
                    } finally {
                        newPage.wUnlock();
                        bufferPoolManager.unpinPage(newPage.getPageId(), true); // 解钉新创建的脏页
                    }
                    bufferPoolManager.unpinPage(currentPageId, true); // 解钉修改了 nextPageId 的旧脏页
                    break;
                }
            } finally {
                currentPage.wUnlock(); // 释放写锁
            }
        }
        if (rid != null) {
            // 如果在事务中，记录插入日志
            if (transaction != null && logManager != null) {
                LogRecord logRecord = new LogRecord(transaction.getTransactionId(), tableInfo.tableName, rid);
                logManager.appendLogRecord(logRecord);
            }
            // 更新索引
            Optional<com.dbms.common.Column> pkColumnOpt = tableInfo.schema.getColumns().stream()
                    .filter(com.dbms.common.Column::isPrimaryKey).findFirst();
            if (pkColumnOpt.isPresent() && pkColumnOpt.get().getColumnType() == Type.INTEGER) {
                int pkIndex = tableInfo.schema.getColumns().indexOf(pkColumnOpt.get());
                tableInfo.index.insert((Integer) tuple.getValue(pkIndex), rid);
            }
            return rid;
        }
        return null;
    }

    @Override
    public void close() {}

    @Override
    public Schema getSchema() { return null; }

    @Override
    public PlanNode getPlanNode() {
        return null;
    }
}